package org.spring.aicloud.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.spring.aicloud.entity.Answer;
import org.spring.aicloud.util.AppVariable;
import org.spring.aicloud.util.idempotent.Idempotent;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Resource
    private ObjectMapper objectMapper;

    private static final String TOPIC = "ai-cloud-test-topic";

    //canal将mysql binlog同步到kafka中的topic
    private static final String CANAL_TOPIC = "aicloud-canal";
    @KafkaListener(topics = {CANAL_TOPIC})
    public void canalListen(String data, Acknowledgment acknowledgment) throws JsonProcessingException {
        HashMap<String,Object> hashMap = objectMapper.readValue(data, HashMap.class);
        //只有在aicloud数据库中answer表才进行处理
        if (!hashMap.isEmpty() && hashMap.get("database").toString().equals("aicloud")
                && hashMap.get("table").toString().equals("answer")){
            //更新redis缓存
            //拿到data下的list
            //注意：list中的内容被转换成了LinkedHashMap
            ArrayList<LinkedHashMap<String,String>> list =
                    (ArrayList<LinkedHashMap<String,String>>)hashMap.get("data");
            String cacheKey = "";
            for (LinkedHashMap<String,String> answer : list){
                cacheKey = AppVariable.getListCacheKey(Long.parseLong(answer.get("uid")),
                        Integer.parseInt(answer.get("model")),
                        Integer.parseInt(answer.get("type"))
                );
                redisTemplate.opsForValue().set(cacheKey,null);
            }
        }
        //手动确认消费
        acknowledgment.acknowledge();
    }

    /**
     * 消息的发送
     */
    @RequestMapping("/send")
    public String send(String msg) {
        if (!StringUtils.hasLength(msg)){
            return "msg is null";
        }
        kafkaTemplate.send(TOPIC, msg);
        return "send";
    }





}
